package prictice.visits;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.apache.zookeeper.*;

import java.io.IOException;
import java.net.InetAddress;
import java.util.Map;

/**
 * @Package prictice.visits
 * @Description:
 * @Author elwyn
 * @Date 2017/9/14 23:48
 * @Email elonyong@163.com
 */
public class PVBolt implements IRichBolt {

    public static final String zkPath = "/lock/storm/pv";
    ZooKeeper zooKeeper = null;
    String lockData = null;

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        try {
            zooKeeper = new ZooKeeper("192.168.128.141:2181,192.168.128.132:2181", 10000, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    System.out.println("watchedEvent = " + watchedEvent.getType());
                }
            });
            while (zooKeeper.getState() != ZooKeeper.States.CONNECTED) {
                Thread.sleep(1000);
            }

            InetAddress inetAddress = InetAddress.getLocalHost();
            lockData = inetAddress.getHostAddress() + ":" + topologyContext.getThisTaskId();
            if (zooKeeper.exists(zkPath, false) == null) {
                zooKeeper.create(zkPath, lockData.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
            try {
                zooKeeper.close();
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
    }

    private String logString = null;
    private String sessionId = null;
    long pv;
    long beginTime = System.currentTimeMillis();
    long endTime = 0;

    @Override
    public void execute(Tuple tuple) {
        try {
            logString = tuple.getString(0);
            if (logString != null) {
                endTime = System.currentTimeMillis();
                sessionId = logString.split("\t")[1];
                if (sessionId != null) {
                    pv++;
                }

            }
            if (endTime - beginTime >= 5 * 1000) {
                System.err.print(lockData + "=================================");
                if (lockData .equals(zooKeeper.getData(zkPath, false, null)) ) {

                    System.err.println(" pv ================ " + pv * 4);
                }
                beginTime = System.currentTimeMillis();
            }

        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void cleanup() {

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
